package tarns_result.others.tmp

import org.apache.spark.SparkConf
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.{DataFrame, Dataset, SparkSession}
import tarns_result.jiaoyu.Match

object UIS_TMP {
  def main(args: Array[String]): Unit = {
    val value_path="hdfs://node01:8020/tmp/InternationalData_source/InternationalData.UIS_Value.json"
    val detail_path="hdfs://node01:8020/tmp/detail/InternationalData.DetailDic.json"
    val match_path = "hdfs://node01:8020/tmp/jiaoyu/match"
    //    val value_path="file:///Users/kelisiduofu/Desktop/UIS/value_uis"
    //    val detail_path="file:///Users/kelisiduofu/Desktop/UIS/detail"
    //    val match_path = "file:///Users/kelisiduofu/Desktop/UIS/match"
    val conf=new SparkConf().setAppName("UIS_TMP")//.setMaster("local[*]")
    val spark=SparkSession.builder().config(conf).getOrCreate()
    val sc=spark.sparkContext
    spark.sql("set spark.sql.caseSensitive=true")
    sc.setLogLevel("WARN")
    val d_v: DataFrame = spark.read.json(detail_path)
    val f_v: DataFrame = spark.read.json(value_path)
    f_v.createTempView("value")
    d_v.createTempView("detail")

    val lines:RDD[String] = sc.textFile(match_path)
    import spark.implicits._
    val rdd_match:RDD[Match]=toMatchRDD(lines).distinct()
    val df: Dataset[Match] = rdd_match.toDS()
    df.createTempView("match")

    val res:DataFrame=spark.sql(
      """
        |select
        |match.*,t3.DetailName
        |from
        |match
        |left join
        |(
        |    select t2.DetailName
        |    from
        |    (
        |        select distinct INDICATOR_ID from value
        |    )t1
        |    left join
        |    (
        |    select distinct DetailId,DetailName from detail
        |    )t2
        |    on t1.INDICATOR_ID=t2.DetailId
        |    where t2.DetailId is not null
        |)t3
        |on match.InternationalIndicator=t3.DetailName
        |where t3.DetailName is not null
        |""".stripMargin)
        res.show()



    res.repartition(1).write.option("delimiter", "\t").csv("hdfs://node01:8020/data/yzw/jiaoyu/UIS_TMP")
      sc.stop()

  }

  def toMatchRDD(lines: RDD[String]):RDD[Match] = {
    val arrays: RDD[Array[String]] = lines.map(line => line.split("\t"))
    val filter_arrays: RDD[Array[String]] = arrays.filter(arr => if (arr.length == 5) true else false)
    //=4的缺少InternationalIndicator，无法关联，直接过滤掉
    filter_arrays.map(arr => Match(arr(4), arr(2), arr(3)))
  }
}

